package cn.itcast.up.model.tools

import cn.itcast.up.model.bean.HBaseMeta
import org.apache.spark.sql.{DataFrame, SQLContext, SaveMode}
import org.apache.spark.sql.sources.{BaseRelation, CreatableRelationProvider, RelationProvider}

/**
  * 自定义Hbase的数据源
  */
class HBaseDataSource extends RelationProvider with CreatableRelationProvider with Serializable {
  /**
    * 自定义数据源的读取数据
    * @param sqlContext
    * @param parameters
    * @return
    */
  override def createRelation(sqlContext: SQLContext, parameters: Map[String, String]): BaseRelation = {
    //将用户传入的参数封装为HBaseMeta对象
    val meta = HBaseMeta(parameters)
    new ReadableRelation(sqlContext, meta)
  }

  /**
    * 自定义数据写入的Source
    * @param sqlContext
    * @param mode 保存模式,追加/覆盖
    * @param parameters
    * @param data 当前要操作的数据集
    * @return
    */
  override def createRelation(sqlContext: SQLContext,
                              mode: SaveMode,
                              parameters: Map[String, String],
                              data: DataFrame): BaseRelation = {
    //将用户传入的参数封装为HBaseMeta对象
    val meta = HBaseMeta(parameters)
    val relation = new WritableRelation(sqlContext, meta, data)
    //调用insert方法,执行数据插入操作
    relation.insert(data, false)
    relation
  }
}
